package com.atguigu.flinksql.day10;

import com.atguigu.datastream.bean.WaterSensor;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * ClassName: Test01
 * Package: com.atguigu.flinksql.day10
 * Description:
 *              从端口读取数据，将数据流转化为动态表，将动态表转为流  追加查询
 * @Author ChenJun
 * @Create 2023/4/18 10:57
 * @Version 1.0
 */
public class Test01 {
    public static void main(String[] args) throws Exception {

        //1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //2. 从端口读取数据并转换为JavaBean
        DataStreamSource<String> streamSource = env.socketTextStream("localhost", 9999);
        SingleOutputStreamOperator<WaterSensor> waterSensorDS = streamSource.map(line -> {
            String[] split = line.split(",");
            return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
        });

        //3. 将流转换为动态表
        //方式一
        Table table = tableEnv.fromDataStream(waterSensorDS);
        tableEnv.createTemporaryView("t1",table);

        //方式二
        tableEnv.createTemporaryView("t2",waterSensorDS);

        //4. 持续查询
        Table table1 = tableEnv.sqlQuery("select * from " + table);
        Table table2 = tableEnv.sqlQuery("select * from t2");

        //5. 直接打印 该打印方式为阻塞模式
//        table1.execute().print();
//        table2.execute().print();

        //6. 将动态表转为流   Row通用类型
        DataStream<Row> rowDataStream = tableEnv.toAppendStream(table1, Row.class);
        DataStream<WaterSensor> waterSensorDataStream = tableEnv.toAppendStream(table1, WaterSensor.class);

        //7. 打印流
        rowDataStream.print("rowDataStream-->");
        waterSensorDataStream.print("waterSensorDataStream-->");

        //8. 执行
        env.execute();

    }
}
